//=====================================================================
//
//  File:        RemoveDuplicates.cs
//  Summary:    This file contains the implementation of the RemoveDuplicates
//                data flow transformation component. This component has a single input
//                and two outputs. The outputs are asynchronous, and the component caches the rows
//                it receives in its input buffer in ProcessInput until it receives notification from 
//                the data flow task that no more rows are coming. Then it sorts the cached rows, and directs duplicate
//                rows to its DuplicateOutput, and distinct rows to its distinct output.
//  Date:        6/15/2004
//
//---------------------------------------------------------------------
//
//  This file is part of the Microsoft SQL Server Code Samples.
//  Copyright (C) Microsoft Corporation.  All rights reserved.
//
//This source code is intended only as a supplement to Microsoft
//Development Tools and/or on-line documentation.  See these other
//materials for detailed information regarding Microsoft code samples.
//
//THIS CODE AND INFORMATION ARE PROVIDED "AS IS" WITHOUT WARRANTY OF //ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO //THE IMPLIED WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A
//PARTICULAR PURPOSE.
//
//===================================================================== 

using System;
using System.Collections;
using System.Data;
using Microsoft.SqlServer.Dts.Pipeline;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using Microsoft.SqlServer.Dts.Runtime;
using Microsoft.SqlServer.Dts.Runtime.Wrapper;
using System.Globalization;

namespace Microsoft.Samples.SqlServer.Dts
{
    /// <summary>
    /// This sample demonstrates a data flow transformation component with asynchronous outputs.
    /// </summary>
    [DtsPipelineComponent(
        DisplayName = "RemoveDuplicatesCS",
        Description = "Finds and removes duplicate rows.",
        IconResource = "Microsoft.Samples.SqlServer.Dts.RemoveDuplicates.ico",
       UITypeName = "Microsoft.Samples.SqlServer.Dts.RemoveDuplicatesUI, RemoveDuplicatesCS, Version=1.0.0.0, Culture=neutral, PublicKeyToken=cc0f57cd8f72ada6",
        ComponentType = ComponentType.Transform)]
    public class RemoveDuplicates : PipelineComponent
    {
        #region Members
        bool fireEventAgain;
        bool cancelEvent;

        Buffer inputBufferCache;
        ColumnInfos colInfoDistinctOutput;
        ColumnInfos colInfoDuplicateOutput;

        /// <summary>
        /// Runtime member. Used to cache the output buffer received in PrimeOutput. This buffer contains
        /// the rows in the output column collection of the meta data.
        /// </summary>
        PipelineBuffer distinctBuffer;
        PipelineBuffer duplicatesBuffer;
        #endregion

        #region Design Time

        #region ProvideComponentProperties
        /// <summary>
        /// Called when the component is initally added to a data flow task. 
        /// Create and configure the input and outputs of the component.
        /// </summary>
        public override void ProvideComponentProperties()
        {
            /// Support resettability.
            /// The base class calls RemoveAllInputsOutputsAndCustomProperties to reset the
            /// component. Used here to highlight the functionality of the base class.
            base.RemoveAllInputsOutputsAndCustomProperties();

            //    Let the base component add the input and output.
            base.ProvideComponentProperties();

            //    Name the input and output, and make the output asynchronous.
            ComponentMetaData.InputCollection[0].Name = "Input";

            IDTSOutput100 distinctOutput = ComponentMetaData.OutputCollection[0];
            distinctOutput.Name = "DistinctOutput";
            distinctOutput.Description = "Distinct rows are directed to this output.";
            distinctOutput.SynchronousInputID = 0;


            ///    Add the duplicate output.
            IDTSOutput100 duplicateOutput = ComponentMetaData.OutputCollection.New();
            duplicateOutput.Name = "DuplicatesOutput";
            duplicateOutput.Description = "Duplicate rows are directed to this output.";


            /// Add a matching custom property to both output objects. This custom property 
            /// is used in PrimeOutput to discover and assign the output buffers to the correct output. 
            /// This is done since the order of the outputs in the collection can not be guaranteed
            /// when the package is initially loaded from an XML document.
            AddIsDistinctCustomProperty(distinctOutput, true);
            AddIsDistinctCustomProperty(duplicateOutput, false);


        }
        #endregion

        #region Validate
        /// <summary>
        /// Called repeatedly when the component is edited in the designer, and once at the beginning of execution.
        /// Verifies the following:
        /// 1. The component has 2 outputs.
        /// 2. The component has 1 input.
        /// 3. There is an output column in each output for each input column.
        /// 4. The base class validation succeeds.
        /// </summary>
        /// <returns></returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes"), CLSCompliant(false)]
        public override DTSValidationStatus Validate()
        {
            if (ComponentMetaData.OutputCollection.Count != 2)
            {
                InternalFireError(Properties.Resources.Exactly2Outputs);
                return DTSValidationStatus.VS_ISCORRUPT;
            }

            if (ComponentMetaData.InputCollection.Count != 1)
            {
                InternalFireError(Properties.Resources.MoreThan1Input);
                return DTSValidationStatus.VS_ISCORRUPT;
            }

            IDTSInput100 input = ComponentMetaData.InputCollection[0];
            IDTSOutput100 distinct = ComponentMetaData.OutputCollection[0];
            IDTSOutput100 duplicate = ComponentMetaData.OutputCollection[1];

            /// Verify that the IsDistinct custom property exists, and that it is true for the 
            /// distinct output.
            if (distinct.CustomPropertyCollection.Count != 0)
            {
                try
                {
                    IDTSCustomProperty100 isDistinct = distinct.CustomPropertyCollection["IsDistinctOutput"];
                    if ((bool)isDistinct.Value == false)
                    {
                        InternalFireError(Properties.Resources.IsDistinctPropertySetIncorrectly);
                        return DTSValidationStatus.VS_NEEDSNEWMETADATA;
                    }
                }
                catch
                {
                    /// The property doesn't exist.
                    InternalFireError(Properties.Resources.IsDistinctPropertyRemoved);
                    return DTSValidationStatus.VS_NEEDSNEWMETADATA;
                }
            }
            else
            {
                InternalFireError(Properties.Resources.IsDistinctPropertyRemoved);
                return DTSValidationStatus.VS_NEEDSNEWMETADATA;
            }
            /// Verify that the IsDistinct custom property exists, and that it is true for the 
            /// distinct output.
            if (duplicate.CustomPropertyCollection.Count != 0)
            {
                try
                {
                    IDTSCustomProperty100 isDistinct = duplicate.CustomPropertyCollection["IsDistinctOutput"];
                    if ((bool)isDistinct.Value == true)
                    {
                        InternalFireError(Properties.Resources.IsDistinctPropertySetIncorrectly);
                        return DTSValidationStatus.VS_NEEDSNEWMETADATA;
                    }
                }
                catch
                {
                    /// The property doesn't exist.
                    InternalFireError(Properties.Resources.IsDistinctPropertyRemoved);
                    return DTSValidationStatus.VS_NEEDSNEWMETADATA;
                }
            }
            else
            {
                InternalFireError(Properties.Resources.IsDistinctPropertyRemoved);
                return DTSValidationStatus.VS_NEEDSNEWMETADATA;
            }
            ///    If there are more output columns than input columns, fail validation
            /// by returning VS_NEEDSNEWMETADATA because ReintializeMetaData fixes this problem.
            if (input.InputColumnCollection.Count != distinct.OutputColumnCollection.Count)
            {
                ComponentMetaData.FireError(0, ComponentMetaData.Name,
                    Properties.Resources.NumberInputColumnsNotMatchOutputColumns
                    + distinct.Name + Properties.Resources.Output, "", 0, out cancelEvent);
                return DTSValidationStatus.VS_NEEDSNEWMETADATA;
            }

            if (input.InputColumnCollection.Count != duplicate.OutputColumnCollection.Count)
            {
                ComponentMetaData.FireError(0, ComponentMetaData.Name,
                    Properties.Resources.NumberInputColumnsNotMatchOutputColumns
                    + duplicate.Name + Properties.Resources.Output, "", 0, out cancelEvent);
                return DTSValidationStatus.VS_NEEDSNEWMETADATA;
            }

            /// If there isn't a matching output column for each input column 
            /// then the component meta data is out of sync. Return VS_NEEDSNEWMETADATA
            /// because ReinitializeMetaData method fixes this problem.
            foreach (IDTSInputColumn100 col in input.InputColumnCollection)
            {
                if (!OutputColumnExists(distinct, col.LineageID))
                {
                    ComponentMetaData.FireError(0, ComponentMetaData.Name, "The input column " + col.Name + " is not found in the " + distinct.Name + " output.", "", 0, out cancelEvent);
                    return DTSValidationStatus.VS_NEEDSNEWMETADATA;
                }

                if (!OutputColumnExists(duplicate, col.LineageID))
                {
                    ComponentMetaData.FireError(0, ComponentMetaData.Name, "The input column " + col.Name + " is not found in the " + duplicate.Name + " output.", "", 0, out cancelEvent);
                    return DTSValidationStatus.VS_NEEDSNEWMETADATA;
                }
            }

            /// Finally, call the base class, which validates that the LineageID of each column in the input collection
            /// matches the LineageID of a column in the VirtualInputColumnCollection. 
            return base.Validate();
        }
        #endregion

        #region SetUsageType
        /// <summary>
        /// The SetUsageType method is called when a column from the virtual input collection is selected for use by the component.
        /// The virtual input collection contains the collection of columns that the upstream component is providing to this component.
        /// If the column DataType is DT_IMAGE it is rejected, since BLOB datatypes are not supported by this component.
        /// Based on the usageType parameter the column is either added, removed, or rejected by the component. 
        /// UsageType.UT_IGNORED - The column was previously added to the input collection. The base class removes the column from the input.
        /// the matching output columns are removed.
        /// UsageType.UT_READONLY - The column is added to the input collection of the component.
        /// UsageType.UT_READWRITE - Since this component does not modify columns this usagetype is rejected.
        /// </summary>
        /// <param name="inputID">The ID of the input the column is mapped to.</param>
        /// <param name="virtualInput">The IDTSVirtualInput100 for the input.</param>
        /// <param name="lineageID">The lineageID of the virtual input column.</param>
        /// <param name="usageType">The usagetype of the column.</param>
        /// <returns>The newly created input column.</returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), CLSCompliant(false)]
        public override IDTSInputColumn100 SetUsageType(int inputID, IDTSVirtualInput100 virtualInput, int lineageID, DTSUsageType usageType)
        {
            if (virtualInput == null)
            {
                throw new ArgumentNullException("virtualInput");
            }

            IDTSVirtualInputColumn100 vCol = virtualInput.VirtualInputColumnCollection.GetVirtualInputColumnByLineageID(lineageID);
            IDTSInputColumn100 col = null;

            /// No support for BLOB image columns.
            if (vCol.DataType == DataType.DT_IMAGE)
            {
                throw new Exception(Properties.Resources.BinaryDataTypesNotSupported);
            }

            ///    If the usageType is UT_IGNORED, then the column is being removed.
            /// So remove it from the outputs also.
            if (usageType == DTSUsageType.UT_IGNORED)
            {
                ///    If the usageType is UT_IGNORED, the base class removes the column
                /// and the returned column is null. 
                col = base.SetUsageType(inputID, virtualInput, lineageID, usageType);

                ///    Remove the matching output columns
                RemoveOutputColumn(ComponentMetaData.OutputCollection[0].ID, vCol.LineageID);
                RemoveOutputColumn(ComponentMetaData.OutputCollection[1].ID, vCol.LineageID);
            }
            else if (usageType == DTSUsageType.UT_READWRITE)
            {
                throw new Exception(Properties.Resources.ReadWriteNotSupported);
            }
            else
            {
                ///    Let the base class add the input column.
                col = base.SetUsageType(inputID, virtualInput, lineageID, usageType);

                /// Add an output column to the distinct and duplicate outputs.
                AddOutputColumn(ComponentMetaData.OutputCollection[0].ID, col);
                AddOutputColumn(ComponentMetaData.OutputCollection[1].ID, col);
            }

            return col;
        }
        #endregion

        #region ReinitializeMetaData
        /// <summary>
        /// Called when VS_NEEDSNEWMETADATA is returned from Validate. 
        /// Reset all of the output columns.
        /// </summary>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")]
        public override void ReinitializeMetaData()
        {
            IDTSOutput100 distinctOutput = ComponentMetaData.OutputCollection[0];
            IDTSOutput100 duplicateOutput = ComponentMetaData.OutputCollection[1];

            /// Verify that the IsDistinct custom property exists, and that it is true for the 
            /// distinct output.
            if (distinctOutput.CustomPropertyCollection.Count != 0)
            {
                try
                {
                    IDTSCustomProperty100 isDistinct = distinctOutput.CustomPropertyCollection["IsDistinctOutput"];
                    if ((bool)isDistinct.Value == false)
                        isDistinct.Value = true;
                }
                catch
                {
                    /// The property doesn't exist, so create it
                    AddIsDistinctCustomProperty(distinctOutput, true);
                }

            }
            else
            {
                AddIsDistinctCustomProperty(distinctOutput, true);
            }
            /// Verify that the IsDistinct custom property exists, and that it is true for the 
            /// distinct output.
            if (duplicateOutput.CustomPropertyCollection.Count != 0)
            {
                try
                {
                    IDTSCustomProperty100 isDistinct = duplicateOutput.CustomPropertyCollection["IsDistinctOutput"];
                    if ((bool)isDistinct.Value == true)
                    {
                        isDistinct.Value = false;
                    }
                }
                catch
                {
                    /// The property doesn't exist, so create it.
                    AddIsDistinctCustomProperty(duplicateOutput, false);
                }
            }
            else
            {
                AddIsDistinctCustomProperty(duplicateOutput, false);
            }

            AddInputColumnsToOutput(ComponentMetaData.OutputCollection[0].ID);
            AddInputColumnsToOutput(ComponentMetaData.OutputCollection[1].ID);
        }
        #endregion

        #region OnInputPathDetached
        /// <summary>
        /// Called when the path connecting the component to an upstream component is deleted. 
        /// Clear the columns from the output collections.
        /// </summary>
        /// <param name="inputID"></param>
        public override void OnInputPathDetached(int inputID)
        {
            base.OnInputPathDetached(inputID);

            ComponentMetaData.OutputCollection[0].OutputColumnCollection.RemoveAll();
            ComponentMetaData.OutputCollection[1].OutputColumnCollection.RemoveAll();
        }
        #endregion

        #region InsertInput
        /// <summary>
        /// Called to add an input to the ComponentMetaData. This component doesn't allow adding inputs,
        /// so an exception is thrown.
        /// </summary>
        /// <param name="insertPlacement"></param>
        /// <param name="inputID"></param>
        /// <returns></returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), CLSCompliant(false)]
        public override IDTSInput100 InsertInput(DTSInsertPlacement insertPlacement, int inputID)
        {
            throw new Exception(Properties.Resources.Component
                + ComponentMetaData.Name
                + Properties.Resources.NotAllowAddingInputs);
        }
        #endregion

        #region InsertOutput
        /// <summary>
        /// Called to add an output to the ComponentMetaData. This component doesn't allow adding outputs,
        /// so an exception is thrown.
        /// </summary>
        /// <param name="insertPlacement"></param>
        /// <param name="outputID"></param>
        /// <returns></returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), CLSCompliant(false)]
        public override IDTSOutput100 InsertOutput(DTSInsertPlacement insertPlacement, int outputID)
        {
            throw new Exception(Properties.Resources.Component
                + ComponentMetaData.Name
                + Properties.Resources.NotAllowAddingOutputs);
        }
        #endregion

        #region InsertOutputColumnAt
        /// <summary>
        /// Called to add an output to the ComponentMetaData. This component doesn't allow adding outputs,
        /// so an exception is thrown.
        /// </summary>
        /// <param name="outputID"></param>
        /// <param name="outputColumnIndex"></param>
        /// <param name="name"></param>
        /// <param name="description"></param>
        /// <returns></returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), CLSCompliant(false)]
        public override IDTSOutputColumn100 InsertOutputColumnAt(int outputID, int outputColumnIndex, string name, string description)
        {
            throw new Exception(Properties.Resources.OutputColumnsCannotBeAdded
                + ComponentMetaData.Name);
        }
        #endregion

        #region SetOutputColumnProperty
        /// <summary>
        /// Called when the custom property on an output column is set. 
        /// Disallow modifications of the InputColumnLineageID custom property by 
        /// throwing an exception.
        /// </summary>
        /// <param name="outputID"></param>
        /// <param name="outputColumnID"></param>
        /// <param name="propertyName"></param>
        /// <param name="propertyValue"></param>
        /// <returns></returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), CLSCompliant(false)]
        public override IDTSCustomProperty100 SetOutputColumnProperty(int outputID, int outputColumnID, string propertyName, object propertyValue)
        {
            if (propertyName == "InputColumnLineageID")
            {
                throw new Exception(Properties.Resources.InputColumnLineageIDPropertyReadOnly);
            }

            return base.SetOutputColumnProperty(outputID, outputColumnID, propertyName, propertyValue);
        }
        #endregion

        #region SetOutputProperty
        /// <summary>
        /// Called when a custom property of an output object is set. The only custom property added and used by this
        /// component is the IsDistinctOutput custom property. It cannot be modified so an exception is thrown.
        /// </summary>
        /// <param name="outputID">The ID of the output.</param>
        /// <param name="propertyName">The name of the property.</param>
        /// <param name="propertyValue">The value to assign to the property.</param>
        /// <returns>The custom property.</returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), CLSCompliant(false)]
        public override IDTSCustomProperty100 SetOutputProperty(int outputID, string propertyName, object propertyValue)
        {
            if (propertyName == "IsDistinctOutput")
            {
                throw new Exception(Properties.Resources.IsDistinctOutputPropertyUsedInternally);
            }

            return base.SetOutputProperty(outputID, propertyName, propertyValue);
        }
        #endregion

        #endregion

        #region Runtime Methods

        #region PreExecute
        /// <summary>
        /// Called prior to PrimeOutput and ProcessInput. 
        ///    Creates the internal buffer and stores the buffer column indexs of the input and output columns
        /// </summary>
        public override void PreExecute()
        {
            colInfoDistinctOutput = new ColumnInfos();
            colInfoDistinctOutput.AddColumnInformation(BufferManager, ComponentMetaData.OutputCollection[0], ComponentMetaData.InputCollection[0]);

            colInfoDuplicateOutput = new ColumnInfos();
            colInfoDuplicateOutput.AddColumnInformation(BufferManager, ComponentMetaData.OutputCollection[1], ComponentMetaData.InputCollection[0]);

            inputBufferCache = new Buffer();
            inputBufferCache.ColumnInfos = colInfoDistinctOutput;
        }
        #endregion

        #region ProcessInput
        /// <summary>
        /// This method is called repeatedly during package execution. It is called each time the data flow task
        /// has a full buffer provided by an upstream component. 
        /// </summary>
        /// <param name="inputID">The ID of the input object of the component.</param>
        /// <param name="buffer">The input buffer.</param>
        public override void ProcessInput(int inputID, PipelineBuffer buffer)
        {
            if (buffer == null)
            {
                throw new ArgumentNullException("buffer");
            }

            //    Have we received the last buffer from the upstream component?
            if (!buffer.EndOfRowset)
            {
                // No. Can we advance the buffer to the next row?
                while (buffer.NextRow())
                {
                    //    Yes, add the row to the internal buffer.
                    inputBufferCache.AddRow(buffer);
                }
            }
            else
            {
                /// Fire sort starting information event.
                DateTime startSort = DateTime.Now;
                ComponentMetaData.FireInformation(0, ComponentMetaData.Name, "Starting sort operation, " + startSort.ToLongTimeString() + ".", "", 0, ref fireEventAgain);
                //    Yes, we've received all of the rows from the upstream component.
                /// Sort the buffer rows.
                inputBufferCache.Sort();

                DateTime sortComplete = DateTime.Now;
                TimeSpan timeToSort = sortComplete.Subtract(startSort);

                /// Fire sort completed information event.
                ComponentMetaData.FireInformation(0, ComponentMetaData.Name,
                    "Internal buffer cache sort complete," + sortComplete.ToLongTimeString()
                    + ", elapsed time was " + timeToSort.TotalSeconds.ToString(CultureInfo.CurrentUICulture)
                    + " seconds.", "", 0, ref fireEventAgain);

                ///    Loop over the rows  in the internal buffer, and add them to the appropriate buffer.
                for (int row = 0; row < inputBufferCache.RowCount; row++)
                {
                    /// If we are not on the first row, and the current row equals the previous row
                    /// then add the row to the duplicates buffer.
                    /// Otherwise, add the row to the distinct buffer.
                    if (row != 0 && inputBufferCache[row].Equals(inputBufferCache[row - 1]))
                    {
                        /// If the buffer exists add the row.
                        if (duplicatesBuffer != null)
                            AddRowToOutputBuffer(duplicatesBuffer, inputBufferCache[row], colInfoDuplicateOutput);
                    }
                    else
                    {
                        ///    If the buffer exists add the row.
                        if (distinctBuffer != null)
                            AddRowToOutputBuffer(distinctBuffer, inputBufferCache[row], colInfoDistinctOutput);
                    }
                }
                /// If an output is not connected, then no output buffer was provided 
                /// by the data flow task during PrimeOutput. Verify the buffer exists before
                /// signaling the data flow task that we are finished adding rows.
                if (distinctBuffer != null)
                    distinctBuffer.SetEndOfRowset();

                if (duplicatesBuffer != null)
                    duplicatesBuffer.SetEndOfRowset();
            }
        }
        #endregion

        #region PrimeOutput
        /// <summary>
        /// Called after PreExecute. The data flow task provides a buffer for each output 
        /// that is connected to a downstream component. Check the location of each output specified in the
        /// outputIDs array, and based on the index of the output cache the output buffer to the appropriate
        /// private member variable.
        /// </summary>
        /// <param name="outputs">The number of outputs.</param>
        /// <param name="outputIDs">An array of output ids.</param>
        /// <param name="buffers">An array of output buffers.</param>
        public override void PrimeOutput(int outputs, int[] outputIDs, PipelineBuffer[] buffers)
        {
            for (int x = 0; x < outputIDs.Length; x++)
            {
                IDTSOutput100 output = ComponentMetaData.OutputCollection.GetObjectByID(outputIDs[x]);

                IDTSCustomProperty100 isDistinctOutput = output.CustomPropertyCollection["IsDistinctOutput"];

                if ((bool)isDistinctOutput.Value)
                    distinctBuffer = buffers[x];
                else
                    duplicatesBuffer = buffers[x];
            }
        }
        #endregion

        #endregion

        #region Helpers
        private void InternalFireError(string Message)
        {
            ComponentMetaData.FireError(0, ComponentMetaData.Name, Message, "", 0, out cancelEvent);
        }
        /// <summary>
        /// For each column in the input collection, add a corresponding column to the 
        /// output column collection of the specified output object.
        /// </summary>
        /// <param name="outputID">The ID of the output to add the columns to.</param>
        private void AddInputColumnsToOutput(int outputID)
        {
            IDTSInput100 input = ComponentMetaData.InputCollection[0];
            IDTSOutput100 output = ComponentMetaData.OutputCollection.GetObjectByID(outputID);

            output.OutputColumnCollection.RemoveAll();

            foreach (IDTSInputColumn100 Col in input.InputColumnCollection)
                AddOutputColumn(outputID, Col);
        }

        /// <summary>
        /// Adds the columns in the provided row object to the provided output buffer.
        /// </summary>
        /// <param name="buffer">The output buffer the rows are added to.</param>
        /// <param name="row">The row object containing the columns that are added to the buffer.</param>
        /// <param name="colInfos">A collection of ColumnInfo objects. There is a one
        /// to one mapping between the ColumnInfo objects and the columns in the row.</param>
        private static void AddRowToOutputBuffer(PipelineBuffer buffer, Row row, ColumnInfos colInfos)
        {
            buffer.AddRow();

            for (int col = 0; col < colInfos.Count; col++)
                buffer[colInfos[col].OutputBufferColumnIndex] = row[col];
        }

        /// <summary>
        /// Creates an output column in the output identified by the OutputID parameter.
        /// Adds a custom property to the output containing the lineageID of the input column. This property is
        /// used during Validate, to verify that each input column has a matching output column.
        /// </summary>
        /// <param name="OutputID">The ID of the output to create the column in.</param>
        /// <param name="inputColumn">The input column used to set the properties of the output column.</param>
        private void AddOutputColumn(int OutputID, IDTSInputColumn100 inputColumn)
        {
            IDTSOutput100 output = ComponentMetaData.OutputCollection.GetObjectByID(OutputID);
            IDTSOutputColumn100 outputColumn = output.OutputColumnCollection.New();
            outputColumn.Name = inputColumn.Name;

            ///    Store the lineageID of the input column in a custom property of the output column.
            IDTSCustomProperty100 inputColLineageID = outputColumn.CustomPropertyCollection.New();
            inputColLineageID.Name = "InputColumnLineageID";
            inputColLineageID.Value = inputColumn.LineageID;

            outputColumn.SetDataTypeProperties(inputColumn.DataType, inputColumn.Length, inputColumn.Precision, inputColumn.Scale, inputColumn.CodePage);
        }
        /// <summary>
        /// Removes the output column that references the input column. The reference is maintained in the 
        /// InputColumnLineageID custom property. 
        /// </summary>
        /// <param name="OutputID">The ID of the output.</param>
        /// <param name="inputColumn">The lineageID of the input column.</param>
        private void RemoveOutputColumn(int OutputID, int LineageID)
        {
            IDTSOutput100 output = ComponentMetaData.OutputCollection.GetObjectByID(OutputID);
            foreach (IDTSOutputColumn100 col in output.OutputColumnCollection)
            {
                IDTSCustomProperty100 inputColumnLineageID = col.CustomPropertyCollection["InputColumnLineageID"];
                if (inputColumnLineageID.Value != null && (int)inputColumnLineageID.Value == LineageID)
                {
                    output.OutputColumnCollection.RemoveObjectByID(col.ID);
                    break;
                }
            }
        }
        /// <summary>
        /// Searches the output column collection for a column containing the provided lineageID
        /// in the InputColumnLineageID custom property.
        /// </summary>
        /// <param name="LineageID">The lineageID of the input column.</param>
        /// <param name="output">The IDTSOutput to search.</param>
        /// <returns>true if found; otherwise false.</returns>
        private static bool OutputColumnExists(IDTSOutput100 output, int LineageID)
        {
            foreach (IDTSOutputColumn100 outColumn in output.OutputColumnCollection)
            {
                /// If the custom property doesn't exist, return false.
                if (outColumn.CustomPropertyCollection.Count != 1)
                    return false;

                if ((int)outColumn.CustomPropertyCollection[0].Value == LineageID)
                    return true;
            }
            return false;
        }

        private static void AddIsDistinctCustomProperty(IDTSOutput100 output, object value)
        {
            IDTSCustomProperty100 isDistinct = output.CustomPropertyCollection.New();
            isDistinct.Name = "IsDistinctOutput";
            isDistinct.Value = value;
        }
        #endregion
    }
}